package edu.vt.ece.rt.quorum.messsages.types;

import java.util.HashSet;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;

import org.jgroups.Address;
import org.jgroups.Message;

import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.dstm.TMProxy;
import edu.vt.ece.rt.quorum.Cluster;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.QMessage;
import edu.vt.ece.rt.quorum.messsages.QMessage.MessageType;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * Commit Request Message to commit DSTM transaction
 * @author Sudhanshu
 */
public class CommitRequest extends QMessage implements Runnable {
	/*
	 * Comparing at local level can be incorrect as version update can be for
	 * this node leading to needless aborts
	 */
	Context cxt; // Context containing the Read-Write Set of objects
	int id; // Requester node of commit request
	Long parentId;

	public CommitRequest(Context c, int i, Long parent, String mId) {
		super(MessageType.CommitRequest);
		cxt = c;
		id = i;
		parentId = parent;
		messageId = mId;
	}


	public static boolean blast(Long txn, Context head) {
		Context hCxt = head;

		if (hCxt.getContextId().equals(txn)) {
			return true;
		} else {
			long t1 = txn / 10000000;
			long t2 = head.getContextId() / 10000000;

			if (t1 == t2) {
				//				return true;
				Context child = ContextDelegator.getContext(txn);
				return Context.isParent(child, head);
			} else {
				return false;
			}
		}
	}

	@Override
	public void run() {
		synchronized (Resource.waitObj) {
			// Remove the Transaction from all PR and PW list
			Cluster cluster;

			Context remoteContext = cxt;
			cxt = ContextDelegator.getContext(cxt.getContextId());
			cxt.setPriority(remoteContext.getPriority());
			//			if (cxt.isRoot()) {
			cluster = ClustersManager.getCluster(id+Names.writeQuorum);
			//			} else {
			//				cluster = ClustersManager.getCluster(id+Names.readQuorum);
			//			}

			Logger.debug("Commit request from " + id + " for tnx "
					+ cxt.getContextId());
			cxt.setParent(parentId); // registers the parent if not already done


			Long txn = cxt.getContextId();
			Resource.rmFrLists(cxt);
			// Create conflicting list
			Set<Context> conflictSet = new HashSet<Context>();

			if (cxt.isAborted()) {
				try {
					Logger.debug("tnx " + txn
							+ " is already aborted, commit response abort");
					cluster.send(
							new Message(getSrcAddr(), new CommitResponse(conflictSet,
									txn, false)));
					Resource.incrementMessages(1);
					// send(new Message CommitResponse(src_addr, null, txn,
					// false));
				} catch (Exception e) {
					e.printStackTrace();
				}
				return;
			}

//			if (cxt.isChild()) {
//				Logger.debug("Start child validation");
////				childValidation(cluster, cxt, remoteContext);
//				return;
//			}





			// Add yourself in conflictSet
			conflictSet.add(cxt);
			HashSet<String> lockedObjects = new HashSet<String>();

			for (QObject wobj : remoteContext.getWriteSet()) {
				
				if (Resource.getQObject(wobj.getId()) == null) {
					System.out.println("new object not found on remote "+ wobj.getId()+ " for "+txn);
				}
				// Check local object status and version comparison
				if (Resource.getQObject(wobj.getId()).isProtected(txn)
						|| Resource.isStaleVersion(wobj)) {
					// True, means abort this transaction
					ContextDelegator.getContext(txn).setAborted();
					try {
						Logger.debug("Object " + wobj.getId()
								+ " is already locked by "
								+ Resource.getQObject(wobj.getId()).getOwner()
								+ " or stale, aborting tnx " + txn);
						cluster
						.send(new Message(getSrcAddr(),
								new CommitResponse(null, txn, false)));
						Resource.incrementMessages(1);
						return;
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						// Unlock all previous taken objects
						unlockObjects(lockedObjects, txn);
					}
				}

				// Add PR and PW to conflicting list
				List<Context> l = Resource.getPRList(wobj.getId());
				//				if (l != null) {
				//					for (Context c : l) {
				//						Logger.debug("PR list context is "+c.getContextId()+ " ");
				//					}
				//				}

				if (l != null)
					conflictSet.addAll(l);
				l = Resource.getPWList(wobj.getId());

				//				if (l!=null) {
				//					for (Context c : l) {
				//						Logger.debug("PW list context is "+c.getContextId()+ " "+c.hashCode());
				//					}
				//				}
				//				
				if (l != null)
					conflictSet.addAll(l);

				Logger.debug("size of Conflict set init is "+conflictSet.size());

				// Check if this transaction is highest priority transaction
				PriorityQueue<Context> prC = new PriorityQueue<Context>(
						conflictSet);
				Context hCxt = prC.peek();
				if (blast(cxt.getContextId(), hCxt)) { // Means still highest
					// priority

					if (cxt.isRoot()
							&& Resource.getQObject(wobj.getId())
							.lockObject(txn)) { // Now try lock
						lockedObjects.add(wobj.getId()); // If you get it add to
						// locked object
						// list
						Logger.debug("Locking for root transaction");
						Logger.debug(txn + " got lock on " + wobj.getId());
						continue;
					}

					if (cxt.isChild()) {
						//						if (wobj.getId().contains("global")) {
						//							// only for the global object on root behalf
						//							if (Resource.getQObject(wobj.getId()).lockObject(cxt.getParentContextId())) {
						//								lockedObjects.add(wobj.getId());
						//							} else {
						//								break;
						//							}
						//						}
						Logger.debug("No locking for child transaction");
						continue;
					} else {
						Logger.debug("Object " + wobj.getId()
								+ " can not be locked, locked by "
								+ wobj.getOwner() + " aborting tnx " + txn);
					}
				}

				Logger.debug(txn + " is not highest Priority for write, highest is "
						+ hCxt.getContextId() + " getting aborted");
				// If not continued must be aborted
				ContextDelegator.getContext(txn).setAborted();
				try {
					cluster.send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					return;
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);
				}

			}
			Logger.debug("size of Conflict set after write set V is "+conflictSet.size());

			// Now check for the read Set
			for (QObject wobj : remoteContext.getReadSet()) {

				if (Resource.getQObject(wobj.getId()).isProtected(txn)
						|| Resource.isStaleVersion(wobj)) {
					// True, means abort this transaction
					ContextDelegator.getContext(txn).setAborted();
					try {
						cluster
						.send(new Message(getSrcAddr(),
								new CommitResponse(null, txn, false)));
						Resource.incrementMessages(1);
						return;
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						// Unlock all previous taken objects
						unlockObjects(lockedObjects, txn);
					}
				}

				// Add PW to conflicting list
				List<Context> l = Resource.getPWList(wobj.getId());
				if (l != null)
					conflictSet.addAll(l);

				// Check if this transaction is highest priority transaction
				PriorityQueue<Context> prC = new PriorityQueue<Context>(
						conflictSet);
				if (blast(cxt.getContextId(), prC.peek())) // Means still
					// highest priority
					continue;

				// If not continued must be aborted
				ContextDelegator.getContext(txn).setAborted();
				try {
					Logger.debug(txn + " is not highest Priority for read, highest is "
							+ prC.peek().getContextId() + ", getting aborted");
					cluster.send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					return;
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);
				}
			}
			Logger.debug("size of Conflict set after read set V is "+conflictSet.size());

			// Check if Context got aborted
			if (ContextDelegator.getContext(txn).isAborted()) {
				try {
					cluster.send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					return;
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);
				}
			}

			// If passed the read and write both validation, it should be
			// committed
			// remove yourself from conflicting list
			conflictSet.remove(cxt);
			Logger.debug("size of Conflict set before is "+conflictSet.size());
			for (Context c : conflictSet) {
				Logger.debug("context is "+c.getContextId());
			}
			while (conflictSet.contains(cxt))
				conflictSet.remove(cxt);
			Logger.debug("size of Conflict set is "+conflictSet.size());
			// Abort all the other transaction in conflict list
			if (ContextDelegator.getContext(txn).isRoot()) {
				for (Context c : conflictSet) {
					if (!blast(cxt.getContextId(), c))
						c.setAborted();
				}
			}

			CommitResponse ca = new CommitResponse(conflictSet, txn, true);
			try {
				Logger.debug("Successfully verified tnx " + txn);
				cluster.send(
						new Message(getSrcAddr(), ca));
				Resource.incrementMessages(cluster.currentView.size());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param lockedObjects
	 */
	private void unlockObjects(HashSet<String> lockedObjects, Long txn) {

		for (String id : lockedObjects) {
			Resource.getQObject(id).unlockObject(txn);
			Logger.debug(txn + " left the lock on " + id);
		}
	}

	/**
	 * Stale Version
	 * 
	 * @param wobj
	 * @param cxt
	 * @return
	 */
	public void childValidation(Cluster cluster,Context cxt, Context remoteContext){
		
		Set<Context> conflictSet = new HashSet<Context>();
		boolean commit = true;
		Long abortTxn = Long.MAX_VALUE;
		Long txn = cxt.getContextId();
		
		Set<QObject> dataSet = remoteContext.getDataSet();
//		Set<QObject> dataSet = new HashSet<QObject>();
//		dataSet.addAll(remoteContext.getReadSet());
//		dataSet.addAll(remoteContext.getWriteSet());
		
		Logger.debug("size of data "+dataSet.size());
		for (QObject obj : dataSet) {
			// Check local object status and version comparison
			Logger.debug("txn : "+txn+" Checking for obj "+ obj.getId() +" clone owner "+obj.getCloneOwner());
			if (Resource.getQObject(obj.getId()).isProtected(txn)
					|| Resource.isStaleVersion(obj)) {
				Logger.debug("txn : "+txn+" Invalid for obj "+ obj.getId());
				// True, means abort this transaction
				ContextDelegator.getContext(txn).setAborted();
				commit = false;
				
				if (obj.getCloneOwner() < abortTxn) { 
					abortTxn = obj.getCloneOwner();
					conflictSet.clear();
					conflictSet.add(ContextDelegator.getContext(abortTxn));
					Logger.debug("txn : "+txn+" updating new clone owner to parent "+abortTxn);
					Resource.rmFrLists(ContextDelegator.getContext(abortTxn));
				}

			}
		}
			if (!commit && abortTxn != Long.MAX_VALUE) {
				Logger.debug("Aborting dummy txn "+abortTxn);
			}
			
			
			
			CommitResponse ca = new CommitResponse(conflictSet, txn,commit);
			try {
				if (commit) {
					Logger.debug("Successfully verified tnx " + txn);
					boolean lockAll = true;
					for (QObject obj : remoteContext.getWriteSet()) {
							if (!Resource.getQObject(obj.getId()).lockObject(remoteContext.getParentContextId())) {
								lockAll = false;
								break;
							}
					}

					if (!lockAll) {
						for (QObject obj : remoteContext.getWriteSet()) {
							Resource.getQObject(obj.getId()).unlockObject(remoteContext.getParentContextId());
						}	
					}

				} else { 
					Logger.debug("Aborting child tnx " + txn +" and parent "+abortTxn);
				}
				cluster.send(
						new Message(getSrcAddr(), ca));
				Resource.incrementMessages(cluster.currentView.size());
			} catch (Exception e) {
				e.printStackTrace();
			}

		
	}
}
